package com.atguigu.datastream.day04;

import com.atguigu.datastream.bean.Event;
import com.atguigu.datastream.day03.Flink_05_Source_userDefaule;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;



/**
 * ClassName: Flink_01_Sink_File
 * Package: com.atguigu.day04
 * Description:
 *                  输出到文件
 * @Author ChenJun
 * @Create 2023/4/10 11:30
 * @Version 1.0
 */
public class Flink_01_Sink_File {
    public static void main(String[] args) throws Exception {

        //1. 创建流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2. 设置并行度
        env.setParallelism(1);

        //3. 读取数据
//        DataStreamSource<String> streamSource = env.readTextFile("input/word.txt");

        DataStreamSource<Event> streamSource = env.addSource(new Flink_05_Source_userDefaule.ClickSource());

        SingleOutputStreamOperator<String> map = streamSource.map(Event::toString);

        StreamingFileSink<String> fileSink = StreamingFileSink.<String>forRowFormat(new Path("./output"),
                        new SimpleStringEncoder<>()).withRollingPolicy(DefaultRollingPolicy.builder()
                        //按照文件大小滚动
//                       .withMaxPartSize(1*1024*1024*1024)
                        .withMaxPartSize(2 * 1024)
                        //按照时间滚动 每隔多久
                        .withRolloverInterval(15 * 60 * 1000)
                        //数据多久每来 滚动
                        .withInactivityInterval(5 * 60 * 1000)
                        .build())
                .build();



        map.addSink(fileSink);

        env.execute();

    }

}
